package com.errol.batchprocessor.batchprocessor;

import com.renjia.entity.AbstractImportBatch;
import com.renjia.service.generic.log.LogService;
import com.renjia.service.generic.middleware.batchprocessor.bo.ThreadPoolAndMissionStatusBO;
import com.renjia.service.generic.middleware.batchprocessor.bo.ThreadPoolStatusBO;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractBatchPayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractPiecePayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractSinglePayload;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.annotation.Async;

/**
 * @author zhang xiao bin
 * @version v1.0
 * @date 2021/02/22
 * @since v5.0.5
 **/
public abstract class AbstractBatchProcessorMonitorImpl<IE extends AbstractImportBatch, AT extends Enum<?>, BE, SE, BP extends AbstractBatchPayload<IE, AT, BE, SE>, PP extends AbstractPiecePayload<SE, SP>, SP extends AbstractSinglePayload<SE>>
        implements BatchProcessorMonitor<IE, AT, BE, SE, BP, PP, SP> {

    private final BatchProcessor<IE, AT, BE, SE, BP, PP, SP> batchProcessor;
    private final LogService logService;

    public AbstractBatchProcessorMonitorImpl(BatchProcessor<IE, AT, BE, SE, BP, PP, SP> batchProcessor,
                                             LogService logService) {
        this.batchProcessor = batchProcessor;
        this.logService = logService;
    }

    @Override
    public void addMissionAsync(@NonNull String batchNo) {
        // TODO: 21/3/8 异常中断重试机制
    }

    @Async
    @Override
    public void addMissionAsync(@NonNull BP batchPayload) {
        try {
            if (shallAddMission(batchPayload)) {
                batchProcessor.startMission(batchPayload);
            } else {
                logService.warn("尝试添加批处理任务，被拦截，batchNo=" + batchPayload.getBatchNo());
            }
        } catch (Exception e) {
            logService.error(e, "尝试添加批处理任务，执行异常，batchNo=" + batchPayload.getBatchNo());
        }
    }

    /**
     * 当前批次是否应该开启，应使用同步机制避免重复开启
     *
     * @param batchPayload 批次数据
     * @return true 开启
     */
    protected abstract boolean shallAddMission(BP batchPayload);

    @Override
    public SP processSingleRecord(BP batchPayload, SP singlePayload) {
        try {
            PP piecePayload = batchProcessor.doSplitBatch(batchPayload).get(0);
            piecePayload.getSinglePayloadList().clear();
            piecePayload.getSinglePayloadList().add(singlePayload);
            if (batchProcessor.doProcessSingleRecord(singlePayload, piecePayload, batchPayload)) {
                return singlePayload;
            }
        } catch (Exception e) {
            logService.error(e);
        } finally {
            batchProcessor.doCleanBatch(batchPayload);
        }
        return null;
    }

    @Override
    public ThreadPoolStatusBO getThreadPoolStatus() {
        return batchProcessor.getThreadPoolStatus();
    }

    @Override
    public ThreadPoolAndMissionStatusBO getMissionStatus() {
        return batchProcessor.getMissionStatus();
    }

    @Override
    public int setCorePoolSize(int targetCorePoolSize) {
        return batchProcessor.setCorePoolSize(targetCorePoolSize);
    }

}
